package com.ndp.fb.rdb.mongo.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mongodb.*;
import com.ndp.fb.constant.MongoConstants;
import com.ndp.fb.model.AdGroupStats;
import com.ndp.fb.model.AdResourceStats;
import com.ndp.fb.model.AdSetStats;
import com.ndp.fb.mongo.api.AdGroupMongoService;
import com.ndp.fb.mongo.api.AdGroupStatsMongoService;
import com.ndp.fb.mongo.api.IMongodbService;
import com.ndp.fb.mongo.model.AdForMongo;
import com.ndp.fb.mongo.model.ImageVersion;
import com.ndp.fb.mongo.model.UpdateAdGroupStatParam;
import com.ndp.fb.util.DateFormatUtil;
import com.ndp.fb.util.ThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * Created by Hermen on 2015/3/5.
 */
@Service
public class AdGroupStatsMongoServiceImpl implements AdGroupStatsMongoService {
    // 可以进行SUM的所有字段
    private static String[] SUM_FIELDS = {"clicks", "impressions", "results", "socialClicks", "socialImpressions", "socialSpent", "socialUniqueClicks", "socialUniqueImpressions", "spent", "uniqueClicks", "uniqueImpressions"};
    private static String[] V2_4_SUM_FIELD = {"appStoreClicks", "clicks", "deeplinkClicks", "impressions", "reach", "results", "socialClicks", "socialImpressions", "socialReach", "socialSpent", "socialUniqueClicks", "socialUniqueImpressions", "spent", "totalActionValue", "totalActions", "uniqueClicks", "uniqueImpressions", "websiteClicks"};

    // 营销目标统计数据列，key为列名，value为mongo中对应的字段
    private static Map<String, String> SUM_FIELDS_FOR_AD_MARKET_TARGET = new HashMap() {
        {
            put("impressions", "impressions");
            put("uniqueImpressions", "uniqueImpressions");
            put("clicks", "clicks");
            put("spent", "spent");
            put("cart", "convertedArray.offsite_conversion.fb_pixel_add_to_cart.actions.1d_click");
            put("purchase", "convertedArray.offsite_conversion.fb_pixel_purchase.actions.1d_click");
            put("salesAmount", "convertedArray.offsite_conversion.fb_pixel_purchase.actionValues.1d_click");
            put("gaSessions", "gaSessions");
            put("gaTransactions", "gaTransactions");
            put("gaTransactionRevenue", "gaTransactionRevenue");
        }
    };

    @Autowired
    private DB mongoDBClient;
    @Autowired
    private DB adHistoryDBClient;
    @Autowired
    private DB mongoDBClientPrimary;

    @Autowired
    private IMongodbService mongodbService;

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private AdGroupMongoService adGroupMongoService;

    /**
     * 根据accountId/adGroupId（二者至少有一个）和day查询统计数据（注意：只返回统计数据，业务数据返回值不可用）
     *
     * @param matchAccountId accountId过滤条件（可选参数）
     * @param matchAdGroupId adGroupId过滤条件（可选参数）
     * @param matchDay       day过滤条件（必选参数）
     * @return
     */
    @Override
    public List<AdGroupStats> queryStatsDataByDay(Long matchAccountId, Long matchAdGroupId, String matchDay) {
        // 构建Match对象
        DBObject matchChildDBObject = new BasicDBObject();
        if (null != matchAccountId) {
            matchChildDBObject.put("accountId", matchAccountId);
        }
        if (null != matchAdGroupId) {
            matchChildDBObject.put("adGroupId", matchAdGroupId);
        }
        matchChildDBObject.put("day", matchDay);
        // 细分字段，all代表没有细分的数据
        matchChildDBObject.put("breakdowns", "all");

        BasicDBObject groupChildDBObject = new BasicDBObject();
        BasicDBObject projectChildDBObject = new BasicDBObject();
        groupChildDBObject.put("_id", new BasicDBObject("adGroupId", "$adGroupId"));
        for (int i = 0; i < SUM_FIELDS.length; i++) {
            addSumFieldToGroupObject(groupChildDBObject, SUM_FIELDS[i]);
            addFieldToTargetObject(projectChildDBObject, SUM_FIELDS[i], 1);
        }
        // cost
        BasicDBObject costDBObject = genConditionDBObject(new BasicDBObject("$eq", new Object[]{"$results", 0}), "$spent", new BasicDBObject("$divide", new Object[]{"$spent", "$results"}));
        projectChildDBObject.put("cost", new BasicDBObject("$cond", costDBObject));

        List<DBObject> aggregateObjects = new ArrayList<>();
        aggregateObjects.add(new BasicDBObject("$match", matchChildDBObject));
        aggregateObjects.add(new BasicDBObject("$group", groupChildDBObject));
        aggregateObjects.add(new BasicDBObject("$project", projectChildDBObject));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<AdGroupStats> list = new ArrayList<>();
        try {
            AdGroupStats adGroupStats = null;
            while (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                dbObject.put("adGroupId", ((DBObject) dbObject.get("_id")).get("adGroupId"));
                adGroupStats = JSONObject.parseObject(JSON.toJSONString(dbObject), AdGroupStats.class);
                list.add(adGroupStats);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return list;
    }

    /**
     * 根据adGroupId和时间范围查询统计数据（注意：只返回统计数据，业务数据返回值不可用）
     *
     * @param matchAdGroupId adGroupId过滤条件（必选参数）
     * @param matchSinceDay  查询开始日过滤条件（必选参数）
     * @param matchUntilDay  查询结束日过滤条件（必选参数）
     * @param returnFields   需要返回的字段过滤条件（可选参数）
     * @return 查询结果列表
     */
    @Override
    public List<AdGroupStats> queryStatsDataByTimeRange(Long matchAdGroupId, String matchSinceDay, String matchUntilDay, String... returnFields) {
        DBObject matchChildDBObject = new BasicDBObject();
        DBObject dayObject = new BasicDBObject();
        dayObject.put("$gte", matchSinceDay);
        dayObject.put("$lte", matchUntilDay);
        matchChildDBObject.put("day", dayObject);
        matchChildDBObject.put("adGroupId", matchAdGroupId);
        // 细分字段，all代表没有细分的数据
        matchChildDBObject.put("breakdowns", "all");
        BasicDBObject groupChildDBObject = new BasicDBObject();

        BasicDBObject idDBObject = new BasicDBObject();
        idDBObject.put("adGroupId", "$adGroupId");
        idDBObject.put("day", "$day");
        groupChildDBObject.put("_id", idDBObject);
        BasicDBObject projectChildDBObject = new BasicDBObject();
        if (0 != returnFields.length) {
            for (int i = 0; i < returnFields.length; i++) {
                String returnField = returnFields[i];
                if (idDBObject.containsField(returnField)) {
                    continue;
                }
                if (Arrays.asList(SUM_FIELDS).contains(returnField)) {
                    addSumFieldToGroupObject(groupChildDBObject, returnField);
                } else {
                    addBusinessFieldToGroupObject(groupChildDBObject, returnField);
                }
                addFieldToTargetObject(projectChildDBObject, returnField, 1);
            }
        } else {
            for (int i = 0; i < SUM_FIELDS.length; i++) {
                addSumFieldToGroupObject(groupChildDBObject, SUM_FIELDS[i]);
                addFieldToTargetObject(projectChildDBObject, SUM_FIELDS[i], 1);
            }
        }
        // cost
        BasicDBObject costDBObject = genConditionDBObject(new BasicDBObject("$eq", new Object[]{"$results", 0}), "$spent", new BasicDBObject("$divide", new Object[]{"$spent", "$results"}));
        projectChildDBObject.put("cost", new BasicDBObject("$cond", costDBObject));

        List<DBObject> aggregateObjects = new ArrayList<>();
        aggregateObjects.add(new BasicDBObject("$match", matchChildDBObject));
        aggregateObjects.add(new BasicDBObject("$group", groupChildDBObject));
        aggregateObjects.add(new BasicDBObject("$project", projectChildDBObject));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<AdGroupStats> listAdGroupStats = new ArrayList<>();
        try {
            while (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                dbObject.put("adGroupId", ((DBObject) dbObject.get("_id")).get("adGroupId"));
                dbObject.put("day", ((DBObject) dbObject.get("_id")).get("day"));
                listAdGroupStats.add(JSONObject.parseObject(JSON.toJSONString(dbObject), AdGroupStats.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }

        return listAdGroupStats;
    }

    /**
     * 以day或accountId或{"day":[campaignId列表]}作为过滤条件，根据campaignId和day进行汇总数据，三者可独立查询（注意：只返回统计数据，业务数据返回值不可用）
     *
     * @param matchDay       查询day过滤条件（可选参数）
     * @param matchAccountId 账号id过滤条件（可选参数）
     * @return
     */
    @Override
    public List<AdSetStats> groupStatsDataByDayAndCampaignId(String matchDay, Long matchAccountId) {
        List<DBObject> aggregateObjects = new ArrayList<>();
        DBObject matchDayObject = new BasicDBObject();
        matchDayObject.put("day", matchDay);
        // 细分字段，all代表没有细分的数据
        matchDayObject.put("breakdowns", "all");
        matchDayObject.put("accountId", matchAccountId);
        aggregateObjects.add(new BasicDBObject("$match", matchDayObject));
        // Group1，根据AdGroupId进行分组，获得业务数据
        DBObject groupObject1 = new BasicDBObject();
        groupObject1.put("_id", "$campaignId");
        for (int i = 0; i < SUM_FIELDS.length; i++) {
            addSumFieldToGroupObject(groupObject1, SUM_FIELDS[i]);
        }
        aggregateObjects.add(new BasicDBObject("$group", groupObject1));
        // project
        BasicDBObject projectObject = new BasicDBObject();
        for (int i = 0; i < SUM_FIELDS.length; i++) {
            addFieldToTargetObject(projectObject, SUM_FIELDS[i], 1);
        }
        // cost
        BasicDBObject costDBObject = genConditionDBObject(new BasicDBObject("$eq", new Object[]{"$results", 0}), "$spent", new BasicDBObject("$divide", new Object[]{"$spent", "$results"}));
        addFieldToTargetObject(projectObject, "cost", new BasicDBObject("$cond", costDBObject));
        aggregateObjects.add(new BasicDBObject("$project", projectObject));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<AdSetStats> listAdGroupStats = new ArrayList<>();
        try {
            DBObject dbObject = null;
            while (cursor.hasNext()) {
                dbObject = cursor.next();
                Long campaignId = (Long) dbObject.get("_id");
                dbObject.put("day", matchDay);
                dbObject.put("campaignId", campaignId);
                dbObject.put("accountId", matchAccountId);
                listAdGroupStats.add(JSONObject.parseObject(JSON.toJSONString(dbObject), AdSetStats.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return listAdGroupStats;
    }

    @Override
    public AdGroupStats getLatestSnap(Long adGroupId, Date triggerPoint) {
        //match
        BasicDBObject matchObj = new BasicDBObject();
        matchObj.put("adGroupId", adGroupId);
        // 细分字段，all代表没有细分的数据
        matchObj.put("breakdowns", "all");
        matchObj.put("day", new BasicDBObject("$lte", DateFormatUtil.formatLong10(triggerPoint)));

        //group
        DBObject groupCdt = new BasicDBObject();
        groupCdt.put("_id", new BasicDBObject("adGroupId", "$adGroupId"));
        groupCdt.put("updateTime", new BasicDBObject("$max", "$dataUpdateTime"));
        appendAllSumFields(groupCdt);

        List<DBObject> pipeline = new ArrayList<>();
        pipeline.add(new BasicDBObject("$match", matchObj));
        pipeline.add(new BasicDBObject("$group", groupCdt));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(pipeline));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(pipeline, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        try {
            if (cursor.hasNext()) {
                DBObject next = cursor.next();
                AdGroupStats result = JSONObject.parseObject(JSON.toJSONString(next), AdGroupStats.class);
                result.setAdGroupId(adGroupId);
                return result;
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return null;
    }

    /**
     * @param adGroupStatsList
     * @discription
     * @author giser.yang
     * 创建时间 2015年7月23日 上午11:31:39
     */
    @Override
    public void batchSaveHistoryAndTemp90Days(List<AdGroupStats> adGroupStatsList) {
        if (checkEmpty(adGroupStatsList)) return;
        // 保存到历史表
        mongodbService.updateHistory(adHistoryDBClient, MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY_HISTORY, JSONArray.toJSONString(adGroupStatsList), new String[]{"adGroupId"});

        // 保存到统计数据90天的临时表
        mongodbService.batchSaveOrUpdateWithoutHistory(mongoDBClient, MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY_90DAYS, JSONArray.toJSONString(adGroupStatsList), new String[]{"mongodbId"}, true);
    }

    private boolean checkEmpty(List<AdGroupStats> adGroupStatsList) {
        if (CollectionUtils.isEmpty(adGroupStatsList)) {
            logger.info("Adgroup统计数据入库列表为空，退出入库操作");
            return true;
        }
        return false;
    }

    /**
     * @param adGroupStatsList
     * @param where
     * @discription
     * @author giser.yang
     * 创建时间 2015年9月22日 下午5:40:22
     */

    @Override
    public void batchSaveByWhereWithoutHistory(
            List<AdGroupStats> adGroupStatsList, String[] where) {
        if (checkEmpty(adGroupStatsList)) return;
        mongodbService.batchSaveOrUpdateWithoutHistory(mongoDBClient, MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY, JSONArray.toJSONString(adGroupStatsList), where, true);
        logger.info("-----向mongoDB：" + MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY + "集合插入数据完成");
    }

    @Override
    public Map<String, Integer> groupByImageHashForActiveAdCount(Set<String> imageHashSet, Set<Long> accountIdSet) {
        // 第一次match，过滤图片及账号
        long currentTimeMillis = System.currentTimeMillis();
        BasicDBObject matchChildObject1 = new BasicDBObject()
                .append("configured_status", "ACTIVE")
                .append("effective_status", "ACTIVE")
                .append("imageInfo.0.imageList.hash", new BasicDBObject("$in", imageHashSet))
                .append("adSet.start_utc_time", new BasicDBObject("$lte", currentTimeMillis))
                .append("$or", new Object[]{new BasicDBObject("adSet.end_utc_time", 0), new BasicDBObject("adSet.end_utc_time", new BasicDBObject("$gte", currentTimeMillis))})
                .append("account_id", new BasicDBObject("$in", accountIdSet));

        // 第一次Project，获取广告当前所使用的图片列表
        BasicDBObject projectChildObject1 = new BasicDBObject()
                .append("adGroupId", "$ad_id")
                .append("imageHashList", createLetObject(new BasicDBObject("imageInfo", new BasicDBObject("$arrayElemAt", new Object[]{"$imageInfo", 0})), "$$imageInfo.imageList.hash"));

        // 第一次unwind，将图片列表分割开来
        String unwindChildObject1 = "$imageHashList";

        // 第二次match，由于轮播图的存在，需要再次筛选出需要统计的数据
        BasicDBObject matchChildObject2 = new BasicDBObject()
                .append("imageHashList", new BasicDBObject("$in", imageHashSet));

        // 第一次group，根据imageHash分组，将相关的adGroupId放到数组中，便于统计数量
        BasicDBObject groupChildObject2 = new BasicDBObject()
                .append("_id", new BasicDBObject("imageHash", "$imageHashList"))
                .append("adGroupId", new BasicDBObject("$addToSet", "$adGroupId"));

        // 第二次project，统计图片数量
        BasicDBObject projectChildObject2 = new BasicDBObject()
                .append("imageHash", "$_id.imageHash")
                .append("adCount", new BasicDBObject("$size", "$adGroupId"));

        // 组合步骤，总共6步
        List<DBObject> aggregateObjects = new ArrayList<>();
        // 第一步，第一次match，过滤图片及账号
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject1));
        // 第二步，第一次Project，获取广告当前所使用的图片列表
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject1));
        // 第三步，第一次unwind，将图片列表分割开来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChildObject1));
        // 第四步，第二次match，由于轮播图的存在，需要再次筛选出需要统计的数据
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject2));
        // 第五步，第一次group，根据imageHash分组，将相关的adGroupId放到数组中，便于统计数量
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject2));
        // 第六步，第二次project，统计图片数量
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject2));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        Map<String, Integer> result = new HashMap<>();
        try {
            DBObject dbObject = null;
            while (cursor.hasNext()) {
                dbObject = cursor.next();
                result.put(dbObject.get("imageHash").toString(), Integer.valueOf(dbObject.get("adCount").toString()));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return result;
    }

    @Override
    public List<AdResourceStats> groupStatsDataByImageHashAndApp(Long matchAccountId, String matchDay) {
        // 第一次Match，过滤统计数据，如果查询新建广告则还要过滤业务数据
        BasicDBObject matchChildObject1 = new BasicDBObject()
                .append("accountId", matchAccountId) // 统计账号过滤
                .append("day", matchDay);

        // 根据adGroupId分组，以便减少Lookup的数据量
        BasicDBObject groupChildObject1 = new BasicDBObject()
                .append("_id", "$adGroupId");
        BasicDBObject statsGroupChildObject = new BasicDBObject()
                .append("_id", "$_id")
                .append("day", "$day")
                .append("breakdowns", "$breakdowns")
                .append("country", "$country")
                .append("age", "$age")
                .append("gender", "$gender");
        groupChildObject1.put("statsData", new BasicDBObject("$addToSet", statsGroupChildObject));

        // lookup，拼接业务数据
        BasicDBObject lookupChildObject = createLookupProject(MongoConstants.TABLE_NAME_ADGROUP, "_id", "ad_id", "adgroup");

        // 将统计数据分离出来
        String unwindChild1Object1 = "$statsData";

        // 第一次Project，主要是根据需要选择后面计算需要用到的业务字段以及统计字段
        BasicDBObject projectChildObject1 = new BasicDBObject()
                .append("imageInfo", new BasicDBObject().append("$filter", new BasicDBObject().append("input", createLetObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup.imageInfo")).append("as", "imageInfo").append("cond", new BasicDBObject().append("$and", new Object[]{new BasicDBObject("$lte", new String[]{"$$imageInfo.startDay", "$statsData.day"}), new BasicDBObject("$gte", new String[]{"$$imageInfo.endDay", "$statsData.day"})}))))
                .append("objective", createLetObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup.campaign.objective"))
                .append("fbConnectionObjectId", createLetObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup.adSet.promoted_object.application_id"))
                .append("_id", "$statsData._id")
                .append("day", "$statsData.day")
                .append("breakdowns", "$statsData.breakdowns")
                .append("country", "$statsData.country")
                .append("age", "$statsData.age")
                .append("gender", "$statsData.gender");

        // 将imageInfo数组分割开来
        String unwindChild1Object2 = "$imageInfo";

        // 第二次Project，主要是为了计算version和imageHashList中的version的差值，便于后面进行过滤
        BasicDBObject projectChildObject2 = new BasicDBObject()
                .append("imageHashList", new BasicDBObject("$filter", new BasicDBObject().append("input", "$imageInfo.imageList").append("as", "imageList").append("cond", new BasicDBObject("$ne", new Object[]{new BasicDBObject("$cond", new Object[]{new BasicDBObject("$or", new Object[]{new BasicDBObject("$eq", new Object[]{"-1", "$$imageList.hash"}), new BasicDBObject("$eq", new Object[]{"0", "$$imageList.hash"})}), null, "$$imageList.hash"}), null}))))
                .append("objective", 1)
                .append("breakdowns", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1)
                .append("fbConnectionObjectId", 1);

        // 将imageHashList数组分割开来
        String unwindChild1Object3 = "$imageHashList";

        // 第一次group，主要是将打散的imageHash再根据统计数据的_id重新合并到一个数组中
        BasicDBObject groupChildObject2 = new BasicDBObject()
                .append("_id", "$_id")
                .append("imageHashList", new BasicDBObject("$addToSet", "$imageHashList.hash"))
                .append("objective", new BasicDBObject("$first", "$objective"))
                .append("breakdowns", new BasicDBObject("$first", "$breakdowns"))
                .append("country", new BasicDBObject("$first", "$country"))
                .append("age", new BasicDBObject("$first", "$age"))
                .append("gender", new BasicDBObject("$first", "$gender"))
                .append("fbConnectionObjectId", new BasicDBObject("$first", "$fbConnectionObjectId"));

        // 第三次Project, 根据每个广告使用的图片个数拆分统计数据
        BasicDBObject projectChildObject3 = new BasicDBObject()
                .append("objective", 1)
                .append("breakdowns", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1)
                .append("fbConnectionObjectId", 1);
        BasicDBObject sizeObject = new BasicDBObject("$size", "$imageHashList");
        // 第一个图片对象，需要加上余数
        BasicDBObject firstImageHashChildObject = new BasicDBObject();
        BasicDBObject firstImageHashChildInputObject = new BasicDBObject("$slice", new Object[]{"$imageHashList", 0, 1});
        BasicDBObject firstImageHashChildInObject = new BasicDBObject();
        firstImageHashChildInObject.put("imageHash", "$$imageHash");
        firstImageHashChildObject.put("input", firstImageHashChildInputObject);
        firstImageHashChildObject.put("as", "imageHash");
        firstImageHashChildObject.put("in", firstImageHashChildInObject);
        // 其它图片对象，不需要加余数
        BasicDBObject otherImageHashChildObject = new BasicDBObject();
        BasicDBObject otherImageHashChildInputObject = new BasicDBObject("$slice", new Object[]{"$imageHashList", 1, sizeObject});
        BasicDBObject otherImageHashChildInObject = new BasicDBObject();
        otherImageHashChildInObject.put("imageHash", "$$imageHash");
        otherImageHashChildObject.put("input", otherImageHashChildInputObject);
        otherImageHashChildObject.put("as", "imageHash");
        otherImageHashChildObject.put("in", otherImageHashChildInObject);
        projectChildObject3.put("imageHashList", new BasicDBObject("$concatArrays", new Object[]{new BasicDBObject("$map", firstImageHashChildObject), new BasicDBObject("$map", otherImageHashChildObject)}));

        // 再次将统计数据根据ImageHash分割开
        String unwindChild1Object4 = "$imageHashList";

        // 第二次group，合并数据计算统计结果
        BasicDBObject groupChildObject3 = new BasicDBObject();
        BasicDBObject idObjectForGroup2 = new BasicDBObject()
                .append("objective", "$objective")
                .append("breakdowns", "$breakdowns")
                .append("country", "$country")
                .append("age", "$age")
                .append("gender", "$gender")
                .append("fbConnectionObjectId", "$fbConnectionObjectId")
                .append("imageHash", "$imageHashList.imageHash");
        groupChildObject3.put("_id", idObjectForGroup2);

        for (String field : SUM_FIELDS) { // 所有的统计字段
            statsGroupChildObject.put(field, "$" + field);
            projectChildObject1.put(field, "$statsData." + field);
            projectChildObject2.put(field, 1);
            groupChildObject2.put(field, new BasicDBObject("$first", "$" + field));
            groupChildObject3.put(field, new BasicDBObject("$sum", "$imageHashList." + field));
            // 余数对象
            BasicDBObject modObject = new BasicDBObject("$mod", new Object[]{"$" + field, sizeObject});
            // 减去余数后整除的对象
            BasicDBObject divideObject = new BasicDBObject("$divide", new Object[]{new BasicDBObject("$subtract", new Object[]{"$" + field, modObject}), sizeObject});
            // 余数加整除对象
            BasicDBObject addModAndDivideObject = new BasicDBObject("$add", new Object[]{divideObject, modObject});
            firstImageHashChildInObject.put(field, addModAndDivideObject);
            otherImageHashChildInObject.put(field, divideObject);
        }

        // 组合步骤，总共12步
        List<DBObject> aggregateObjects = new ArrayList<>();
        // 第1步，match，过滤统计数据，如果查询新建广告则还要过滤业务数据
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject1));
        // 第2步，根据adGroupId分组，以便减少Lookup的数据量
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject1));
        // 第3步，lookup，拼接业务数据
        aggregateObjects.add(new BasicDBObject("$lookup", lookupChildObject));
        // 第4步，unwind，将统计数据分离出来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object1));
        // 第5步，project，主要是根据需要选择后面计算需要用到的业务字段以及统计字段
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject1));
        // 第6步，unwind，将imageInfo数组分割开来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object2));
        // 第7步，project，主要是为了计算version和imageHashList中的version的差值，便于后面进行过滤
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject2));
        // 第8步，unwind，将imageHashList数组分割开来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object3));
        // 第9步，group，主要是将打散的imageHash再根据统计数据的_id重新合并到一个数组中
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject2));
        // 第10步，project，根据每个广告使用的图片个数拆分统计数据
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject3));
        // 第11步，unwind，再次将统计数据根据ImageHash分割开
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object4));
        // 第12步，group，合并数据计算统计结果
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject3));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<AdResourceStats> adResourceStatsList = new ArrayList<>();
        try {
            DBObject dbObject = null;
            while (cursor.hasNext()) {
                dbObject = cursor.next();
                Map idResult = (Map) dbObject.get("_id");
                dbObject.putAll(idResult);
                adResourceStatsList.add(JSONObject.parseObject(JSON.toJSONString(dbObject), AdResourceStats.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return adResourceStatsList;
    }

    @Override
    public List<AdResourceStats> groupStatsDataByImageHashAndAppForNewAd(Long matchAdGroupId, String startDay, String endDay, List<ImageVersion> imageInfo) {
        // 第一次Match，根据day和adGroupId过滤统计数据
        BasicDBObject matchChildObject1 = new BasicDBObject()
                .append("adGroupId", matchAdGroupId) // 统计账号过滤
                .append("day", new BasicDBObject().append("$gte", startDay).append("$lte", endDay));

        // 第一次Project，主要是根据需要选择后面计算需要用到的业务字段以及统计字段
        BasicDBObject projectChildObject1 = new BasicDBObject()
                .append("imageInfo", new BasicDBObject().append("$filter", new BasicDBObject().append("input", new BasicDBObject("$literal", com.mongodb.util.JSON.parse(JSON.toJSONString(imageInfo)))).append("as", "imageInfo").append("cond", new BasicDBObject().append("$and", new Object[]{new BasicDBObject("$lte", new String[]{"$$imageInfo.startDay", "$day"}), new BasicDBObject("$gte", new String[]{"$$imageInfo.endDay", "$day"})}))))
                .append("day", "$day")
                .append("breakdowns", "$breakdowns")
                .append("country", "$country")
                .append("age", "$age")
                .append("gender", "$gender");

        // 将imageInfo数组分割开来
        String unwindChild1Object1 = "$imageInfo";

        // 第二次Project，主要是为了计算version和imageHashList中的version的差值，便于后面进行过滤
        BasicDBObject projectChildObject2 = new BasicDBObject()
                .append("imageHashList", new BasicDBObject("$filter", new BasicDBObject().append("input", "$imageInfo.imageList").append("as", "imageList").append("cond", new BasicDBObject("$ne", new Object[]{new BasicDBObject("$cond", new Object[]{new BasicDBObject("$or", new Object[]{new BasicDBObject("$eq", new Object[]{"-1", "$$imageList.hash"}), new BasicDBObject("$eq", new Object[]{"0", "$$imageList.hash"})}), null, "$$imageList.hash"}), null}))))
                .append("breakdowns", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1);

        // 将imageHashList数组分割开来
        String unwindChild1Object2 = "$imageHashList";

        // 第一次group，主要是将打散的imageHash再根据统计数据的_id重新合并到一个数组中
        BasicDBObject groupChildObject1 = new BasicDBObject()
                .append("_id", "$_id")
                .append("imageHashList", new BasicDBObject("$addToSet", "$imageHashList.hash"))
                .append("breakdowns", new BasicDBObject("$first", "$breakdowns"))
                .append("country", new BasicDBObject("$first", "$country"))
                .append("age", new BasicDBObject("$first", "$age"))
                .append("gender", new BasicDBObject("$first", "$gender"));

        // 第三次Project, 根据每个广告使用的图片个数拆分统计数据
        BasicDBObject projectChildObject3 = new BasicDBObject()
                .append("breakdowns", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1);
        BasicDBObject sizeObject = new BasicDBObject("$size", "$imageHashList");
        // 第一个图片对象，需要加上余数
        BasicDBObject firstImageHashChildObject = new BasicDBObject();
        BasicDBObject firstImageHashChildInputObject = new BasicDBObject("$slice", new Object[]{"$imageHashList", 0, 1});
        BasicDBObject firstImageHashChildInObject = new BasicDBObject();
        firstImageHashChildInObject.put("imageHash", "$$imageHash");
        firstImageHashChildObject.put("input", firstImageHashChildInputObject);
        firstImageHashChildObject.put("as", "imageHash");
        firstImageHashChildObject.put("in", firstImageHashChildInObject);
        // 其它图片对象，不需要加余数
        BasicDBObject otherImageHashChildObject = new BasicDBObject();
        BasicDBObject otherImageHashChildInputObject = new BasicDBObject("$slice", new Object[]{"$imageHashList", 1, sizeObject});
        BasicDBObject otherImageHashChildInObject = new BasicDBObject();
        otherImageHashChildInObject.put("imageHash", "$$imageHash");
        otherImageHashChildObject.put("input", otherImageHashChildInputObject);
        otherImageHashChildObject.put("as", "imageHash");
        otherImageHashChildObject.put("in", otherImageHashChildInObject);
        projectChildObject3.put("imageHashList", new BasicDBObject("$concatArrays", new Object[]{new BasicDBObject("$map", firstImageHashChildObject), new BasicDBObject("$map", otherImageHashChildObject)}));

        // 再次将统计数据根据ImageHash分割开
        String unwindChild1Object3 = "$imageHashList";

        // 第二次group，合并数据计算统计结果
        BasicDBObject groupChildObject2 = new BasicDBObject();
        BasicDBObject idObjectForGroup2 = new BasicDBObject()
                .append("objective", "$objective")
                .append("breakdowns", "$breakdowns")
                .append("country", "$country")
                .append("age", "$age")
                .append("gender", "$gender")
                .append("fbConnectionObjectId", "$fbConnectionObjectId")
                .append("imageHash", "$imageHashList.imageHash");
        groupChildObject2.put("_id", idObjectForGroup2);

        for (String field : SUM_FIELDS) { // 所有的统计字段
            projectChildObject1.put(field, 1);
            projectChildObject2.put(field, 1);
            groupChildObject1.put(field, new BasicDBObject("$first", "$" + field));
            groupChildObject2.put(field, new BasicDBObject("$sum", "$imageHashList." + field));
            // 余数对象
            BasicDBObject modObject = new BasicDBObject("$mod", new Object[]{"$" + field, sizeObject});
            // 减去余数后整除的对象
            BasicDBObject divideObject = new BasicDBObject("$divide", new Object[]{new BasicDBObject("$subtract", new Object[]{"$" + field, modObject}), sizeObject});
            // 余数加整除对象
            BasicDBObject addModAndDivideObject = new BasicDBObject("$add", new Object[]{divideObject, modObject});
            firstImageHashChildInObject.put(field, addModAndDivideObject);
            otherImageHashChildInObject.put(field, divideObject);
        }

        // 组合步骤，总共13步
        List<DBObject> aggregateObjects = new ArrayList<>();
        // 第1步，match，过滤统计数据，如果查询新建广告则还要过滤业务数据
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject1));
        // 第2步，project，主要是根据需要选择后面计算需要用到的业务字段以及统计字段
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject1));
        // 第3步，unwind，将imageInfo数组分割开来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object1));
        // 第4步，project，主要是为了计算version和imageHashList中的version的差值，便于后面进行过滤
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject2));
        // 第5步，unwind，将imageHashList数组分割开来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object2));
        // 第6步，group，主要是将打散的imageHash再根据统计数据的_id重新合并到一个数组中
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject1));
        // 第7步，project，根据每个广告使用的图片个数拆分统计数据
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject3));
        // 第8步，unwind，再次将统计数据根据ImageHash分割开
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object3));
        // 第9步，group，合并数据计算统计结果
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject2));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        List<AdResourceStats> adResourceStatsList = new ArrayList<>();
        try {
            DBObject dbObject = null;
            while (cursor.hasNext()) {
                dbObject = cursor.next();
                Map idResult = (Map) dbObject.get("_id");
                dbObject.putAll(idResult);
                adResourceStatsList.add(JSONObject.parseObject(JSON.toJSONString(dbObject), AdResourceStats.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return adResourceStatsList;
    }

    @Override
    public List<AdGroupStats> getDetail(Long accountId, Long adGroupId, String day, String breakdowns, Integer pIndex, Integer pSize) {
        // 第一次Match，过滤统计数据
        BasicDBObject matchChildObject1 = new BasicDBObject()
                .append("day", day) // 统计周期过滤
                .append("breakdowns", breakdowns);  // 细分字段过滤
        if (null != accountId) {
            matchChildObject1.append("accountId", accountId); // 统计账号过滤
        }
        if (null != adGroupId) {
            matchChildObject1.append("adGroupId", adGroupId); // 广告过滤
        }

        // 根据adGroupId分组，以便减少Lookup的数据量
        BasicDBObject groupChildObject1 = new BasicDBObject()
                .append("_id", "$adGroupId");
        BasicDBObject statsGroupChildObject = new BasicDBObject()
                .append("_id", "$_id")
                .append("adGroupId", "$adGroupId")
                .append("accountId", "$accountId")
                .append("day", "$day")
                .append("country", "$country")
                .append("age", "$age")
                .append("gender", "$gender");
        groupChildObject1.put("statsData", new BasicDBObject("$addToSet", statsGroupChildObject));

        // lookup，拼接业务数据
        BasicDBObject lookupChildObject = createLookupProject(MongoConstants.TABLE_NAME_ADGROUP, "_id", "ad_id", "adgroup");

        // unwind，拼接到业务数据后将之前group的统计数据再分开
        String unwindChildObject1 = "$statsData";

        // 第一次Project，主要是根据需要选择后面计算需要用到的业务字段以及统计字段
        BasicDBObject projectChildObject1 = new BasicDBObject()
                .append("_id", "$statsData._id")
                .append("adGroupId", "$statsData.adGroupId")
                .append("accountId", "$statsData.accountId")
                .append("country", "$statsData.country")
                .append("age", "$statsData.age")
                .append("gender", "$statsData.gender")
                .append("imageInfo", new BasicDBObject().append("$filter", new BasicDBObject().append("input", createLetObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup.imageInfo")).append("as", "imageInfo").append("cond", new BasicDBObject().append("$and", new Object[]{new BasicDBObject("$lte", new String[]{"$$imageInfo.startDay", "$statsData.day"}), new BasicDBObject("$gte", new String[]{"$$imageInfo.endDay", "$statsData.day"})}))));

        // 将imageInfo数组分割开来
        BasicDBObject unwindChildObject2 = new BasicDBObject()
                .append("path", "$imageInfo")
                .append("preserveNullAndEmptyArrays", true);

        // 第二次Project，主要是为了计算version和imageHashList中的version的差值，便于后面进行过滤
        BasicDBObject projectChildObject2 = new BasicDBObject()
                .append("adGroupId", 1)
                .append("accountId", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1)
                .append("imageHashList", new BasicDBObject("$filter", new BasicDBObject().append("input", "$imageInfo.imageList").append("as", "imageList").append("cond", new BasicDBObject("$ne", new Object[]{new BasicDBObject("$cond", new Object[]{new BasicDBObject("$or", new Object[]{new BasicDBObject("$eq", new Object[]{"-1", "$$imageList.hash"}), new BasicDBObject("$eq", new Object[]{"0", "$$imageList.hash"})}), null, "$$imageList.hash"}), null}))));

        // 将imageHashList数组分割开来
        BasicDBObject unwindChild1Object3 = new BasicDBObject()
                .append("path", "$imageHashList")
                .append("preserveNullAndEmptyArrays", true);

        // 第一次group，主要是将打散的imageHash再根据统计数据的_id重新合并到一个数组中
        BasicDBObject groupChildObject2 = new BasicDBObject()
                .append("_id", "$_id")
                .append("adGroupId", new BasicDBObject("$first", "$adGroupId"))
                .append("accountId", new BasicDBObject("$first", "$accountId"))
                .append("country", new BasicDBObject("$first", "$country"))
                .append("age", new BasicDBObject("$first", "$age"))
                .append("gender", new BasicDBObject("$first", "$gender"))
                .append("imageHashList", new BasicDBObject("$addToSet", "$imageHashList.hash"));

        // 第三次Project，给未找到图片的统计数据拼接一个空数组，便于后面进行计算
        BasicDBObject projectChildObject3 = new BasicDBObject()
                .append("adGroupId", 1)
                .append("accountId", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1)
                .append("imageHashList", new BasicDBObject().append("$cond", new Object[]{new BasicDBObject().append("$eq", new Object[]{null, new BasicDBObject("$concat", new Object[]{new BasicDBObject("$arrayElemAt", new Object[]{"$imageHashList", 0}), ""})}), new String[]{""}, "$imageHashList"}));

        // 第四次Project, 根据每个广告使用的图片个数拆分统计数据
        BasicDBObject projectChildObject4 = new BasicDBObject()
                .append("accountId", 1)
                .append("adGroupId", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1);
        BasicDBObject sizeObject = new BasicDBObject("$size", "$imageHashList");
        // 第一个图片对象，需要加上余数
        BasicDBObject firstImageHashChildObject = new BasicDBObject();
        BasicDBObject firstImageHashChildInputObject = new BasicDBObject("$slice", new Object[]{"$imageHashList", 0, 1});
        BasicDBObject firstImageHashChildInObject = new BasicDBObject();
        firstImageHashChildInObject.put("imageHash", "$$imageHash");
        firstImageHashChildObject.put("input", firstImageHashChildInputObject);
        firstImageHashChildObject.put("as", "imageHash");
        firstImageHashChildObject.put("in", firstImageHashChildInObject);
        // 其它图片对象，不需要加余数
        BasicDBObject otherImageHashChildObject = new BasicDBObject();
        BasicDBObject otherImageHashChildInputObject = new BasicDBObject("$slice", new Object[]{"$imageHashList", 1, sizeObject});
        BasicDBObject otherImageHashChildInObject = new BasicDBObject();
        otherImageHashChildInObject.put("imageHash", "$$imageHash");
        otherImageHashChildObject.put("input", otherImageHashChildInputObject);
        otherImageHashChildObject.put("as", "imageHash");
        otherImageHashChildObject.put("in", otherImageHashChildInObject);
        projectChildObject4.put("imageHashList", new BasicDBObject("$concatArrays", new Object[]{new BasicDBObject("$map", firstImageHashChildObject), new BasicDBObject("$map", otherImageHashChildObject)}));

        // 再次将统计数据根据ImageHash分割开
        BasicDBObject unwindChild1Object4 = new BasicDBObject()
                .append("path", "$imageHashList")
                .append("preserveNullAndEmptyArrays", true);

        // 第五次Project, 将子对象的数据提到顶层显示
        BasicDBObject projectChildObject5 = new BasicDBObject()
                .append("accountId", 1)
                .append("adGroupId", 1)
                .append("country", 1)
                .append("age", 1)
                .append("gender", 1)
                .append("imageHash", "$imageHashList.imageHash");

        for (String field : V2_4_SUM_FIELD) { // 所有的统计字段
            statsGroupChildObject.put(field, "$" + field);
            projectChildObject1.put(field, "$statsData." + field);
            projectChildObject2.put(field, 1);
            projectChildObject3.put(field, 1);
            groupChildObject2.put(field, new BasicDBObject("$first", "$" + field));
            // 余数对象
            BasicDBObject modObject = new BasicDBObject("$mod", new Object[]{"$" + field, sizeObject});
            // 减去余数后整除的对象
            BasicDBObject divideObject = new BasicDBObject("$divide", new Object[]{new BasicDBObject("$subtract", new Object[]{"$" + field, modObject}), sizeObject});
            // 余数加整除对象
            BasicDBObject addModAndDivideObject = new BasicDBObject("$add", new Object[]{divideObject, modObject});
            firstImageHashChildInObject.put(field, addModAndDivideObject);
            otherImageHashChildInObject.put(field, divideObject);
            projectChildObject5.put(field, "$imageHashList." + field);
        }

        int skipChildObject = (pIndex - 1) * pSize;

        // 组合步骤，总共15步
        List<DBObject> aggregateObjects = new ArrayList<>();
        // 第一步，match，过滤统计数据
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject1));
        // 第二步，group，根据adGroupId分组，以便减少Lookup的数据量
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject1));
        // 第三步，lookup，拼接业务数据
        aggregateObjects.add(new BasicDBObject("$lookup", lookupChildObject));
        // 第四步，unwind，拼接到业务数据后将之前group的统计数据再分开
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChildObject1));
        // 第五步，project，主要是根据需要选择后面计算需要用到的业务字段以及统计字段
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject1));
        // 第六步，unwind，将imageInfo数组分割开来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChildObject2));
        // 第七步，project，主要是为了计算version和imageHashList中的version的差值，便于后面进行过滤
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject2));
        // 第八步，unwind，将imageHashList数组分割开来
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object3));
        // 第九步，group，主要是将打散的imageHash再根据统计数据的_id重新合并到一个数组中
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject2));
        // 第十步，project，给未找到图片的统计数据拼接一个空数组，便于后面进行计算
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject3));
        // 第十一步，project，根据每个广告使用的图片个数拆分统计数据
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject4));
        // 第十二步，unwind，将统计结果数组分开
        aggregateObjects.add(new BasicDBObject("$unwind", unwindChild1Object4));
        // 第十三步，project，将子对象的数据提到顶层显示
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject5));
        // 第十四步，skip，分页
        aggregateObjects.add(new BasicDBObject("$skip", skipChildObject));
        // 第十三步，limit，分页
        aggregateObjects.add(new BasicDBObject("$limit", pSize));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));

        List<AdGroupStats> result = new ArrayList<>();
        DBObject dbObject = null;
        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = null;
        try {
            //出错重试3次   todo refactor to business
            for (int i = 0; i < 3; i++) {
                try {
                    cursor = mongoDBClientPrimary.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateObjects, aggregationOptions);
                    logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
                    break;
                } catch (MongoException e) {
                    logger.warn("链接mongo错误，第" + i + "重试", e);
                }
            }
            while (cursor.hasNext()) {
                dbObject = cursor.next();
                result.add(JSONObject.parseObject(JSON.toJSONString(dbObject), AdGroupStats.class));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (null != cursor) {
                cursor.close();
            }
        }
        return result;
    }

    /**
     * 根据统计数据表的adGroupId字段和业务数据表的ad_id字段拼接两个表，做关联查询
     *
     * @return 拼接对象
     */
    private BasicDBObject createLookupProject(String from, String localField, String foreignField, String as) {
        // lookup，拼接业务数据
        return new BasicDBObject()
                .append("from", from)
                .append("localField", localField)
                .append("foreignField", foreignField)
                .append("as", as);
    }

    /**
     * 往给定的GroupDBObject中追加业务的字段（取第一条数据的字段）
     *
     * @param groupDBObject 需要追加的GroupDBObject
     * @param fieldName     追加字段名称
     */
    private void addBusinessFieldToGroupObject(DBObject groupDBObject, String fieldName) {
        groupDBObject.put(fieldName, "$" + fieldName);
    }

    /**
     * 往给定的GroupDBObject中追加sum的字段
     *
     * @param groupDBObject 需要追加的GroupDBObject
     * @param fieldName     追加字段名称
     */
    private void addSumFieldToGroupObject(DBObject groupDBObject, String fieldName) {
        groupDBObject.put(fieldName, new BasicDBObject("$sum", "$" + fieldName));
    }

    /**
     * 往给定的targetObject中追加字段
     *
     * @param targetObject 需要追加的ProjectDBObject
     * @param fieldName    追加字段名称
     * @param value        追加字段对应的值
     */
    private void addFieldToTargetObject(DBObject targetObject, String fieldName, Object value) {
        targetObject.put(fieldName, value);
    }

    /**
     * 生成一个条件表达式DBObject
     *
     * @param ifObject   ifObject
     * @param thenObject thenObject
     * @param elseObject elseObject
     * @return
     */
    private BasicDBObject genConditionDBObject(Object ifObject, Object thenObject, Object elseObject) {
        BasicDBObject condDBObject = new BasicDBObject();
        condDBObject.put("if", ifObject);
        condDBObject.put("then", thenObject);
        condDBObject.put("else", elseObject);
        return condDBObject;
    }

    private void appendAllSumFields(DBObject groupCdt) {
        for (int i = 0; i < SUM_FIELDS.length; i++) {
            addSumFieldToGroupObject(groupCdt, SUM_FIELDS[i]);
        }
    }

    private BasicDBObject createLetObject(Object varsObject, Object inObject) {
        return new BasicDBObject("$let", new BasicDBObject().append("vars", varsObject).append("in", inObject));
    }

    @Override
    public Map<Long, Long> getAdSpentGroupByAccountId(String day) {
        List<DBObject> pipeline = new ArrayList<>();
        DBObject match = new BasicDBObject("day", day).append("breakdowns", "all");
        DBObject group = new BasicDBObject("_id", "$account_id");
        group.put("total", new BasicDBObject("$sum", "$spent"));
        pipeline.add(new BasicDBObject("$match", match));
        pipeline.add(new BasicDBObject("$group", group));
        Map<Long, Long> result = new HashMap<>();
        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP).aggregate(pipeline, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        try {
            if (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                if (dbObject != null && dbObject.get("total") != null) {
                    result.put(Long.valueOf(dbObject.get("_id").toString()), Long.valueOf(dbObject.get("total").toString()));
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return result;
    }

    @Override
    public JSONObject groupByAdMarketTarget(List<Long> adIdList, String today, String yesterday) {
        // 过滤，根据adGroupId进行过滤
        BasicDBObject matchChildObject1 = new BasicDBObject("adGroupId", new BasicDBObject("$in", adIdList));

        // 合并计算
        BasicDBObject groupChildObject1 = new BasicDBObject("_id", null);
        SUM_FIELDS_FOR_AD_MARKET_TARGET.forEach((key, value) -> {
            groupChildObject1.append(key, new BasicDBObject("$sum", "$" + value));
            groupChildObject1.append("yesterday" + key.replaceFirst(key.substring(0, 1), key.substring(0, 1).toUpperCase()),
                    new BasicDBObject("$sum", new BasicDBObject()
                            .append("$cond", new BasicDBObject()
                                    .append("if", new BasicDBObject("$eq", new String[]{"$day", yesterday}))
                                    .append("then", "$" + value)
                                    .append("else", 0))));
            groupChildObject1.append("today" + key.replaceFirst(key.substring(0, 1), key.substring(0, 1).toUpperCase()),
                    new BasicDBObject("$sum", new BasicDBObject()
                            .append("$cond", new BasicDBObject()
                                    .append("if", new BasicDBObject("$eq", new String[]{"$day", today}))
                                    .append("then", "$" + value)
                                    .append("else", 0))));
        });
        List<DBObject> aggregateObjects = new ArrayList<DBObject>();
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject1));
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject1));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));
        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY_90DAYS).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        JSONObject result = null;
        try {
            if (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                result = JSON.parseObject(JSON.toJSONString(dbObject));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return result;
    }

    @Override
    public void updateStatsDataByDay(List<UpdateAdGroupStatParam> updateAdGroupStatParams, List<Long> actIds) {
        ThreadPool.execute(() -> {
            for (UpdateAdGroupStatParam updateAdGroupStatParam : updateAdGroupStatParams) {
            	Long ad_id = updateAdGroupStatParam.getAd_id();
            	if(ad_id==null){
            		 BasicDBObject nameFinder = new BasicDBObject();
                     nameFinder.put("$regex", updateAdGroupStatParam.getAdName());
                     nameFinder.put("$options", "i");
                     BasicDBObject matchChildObject1 = new BasicDBObject("name", nameFinder);

                     BasicDBObject actFinder = new BasicDBObject();
                     actFinder.put("$in", actIds);
                     matchChildObject1.put("account_id", actFinder);
                     DBObject one = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP).findOne(matchChildObject1);
                     if (one != null) {
                         ad_id = (Long)one.get("ad_id");
                     }
            	}
                if (ad_id != null) {
                    BasicDBObject query = new BasicDBObject();
                    query.put("adGroupId", ad_id);
                    query.put("day", updateAdGroupStatParam.getStartDate());

                    BasicDBObject upper = new BasicDBObject();

                    upper.put("$set", updateAdGroupStatParam.getValueMap());
                    WriteResult update = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY_90DAYS).update(query, upper, false, false);
                    System.out.println("query>>>>>>>>" + query + ";>>>>>>>>>" + update);
                }
            }
        });
    }

    @Override
    public JSONArray groupByDay(String adMarketTargetId, Long adId, String startDay, String endDay) {
        Set<Long> adIds = new HashSet<>();
        if (null != adId && 0 != adId) {
            adIds.add(adId);
        }
        if (null != adMarketTargetId) { // 如果adMarketTargetId不为空的话，那么先要查到所有的ad
            List<AdForMongo> ads = adGroupMongoService.findAdByAdMarketTargetId(adMarketTargetId, "ad_id");
            if (!CollectionUtils.isEmpty(ads)) {
                ads.forEach(ad -> {
                    adIds.add(ad.getAd_id());
                });
            }
        }

        // 根据查询条件进行过滤
        BasicDBObject matchChildObject1 = new BasicDBObject()
                .append("adGroupId", new BasicDBObject("$in", adIds))
                .append("day", new BasicDBObject()
                        .append("$gte", startDay)
                        .append("$lte", endDay));

        // 按天分组统计数据
        BasicDBObject groupChildObject1 = new BasicDBObject()
                .append("_id", "$day");

        // 计算ctr,cpc等数据
        BasicDBObject projectChildObject1 = new BasicDBObject()
                .append("_id", 0)
                .append("day", "$_id");
        SUM_FIELDS_FOR_AD_MARKET_TARGET.forEach((key, value) -> {
            groupChildObject1.append(key, new BasicDBObject("$sum", "$" + value));
            projectChildObject1.append(key, 1);
        });

        // ctr
        projectChildObject1.append("ctr", createDivideObject("$clicks", "$impressions", 0L));
        // cpc
        projectChildObject1.append("cpc", createDivideObject("$spent", "$clicks", "$spent"));
        // purchaseR
        projectChildObject1.append("purchaseR", createDivideObject("$purchase", "$cart", 0L));
        // cps
        projectChildObject1.append("cps", createDivideObject("$spent", "$purchase", "$spent"));
        // roi
        projectChildObject1.append("roi", createDivideObject("$salesAmount", "$spent", 0L));
        // gaRoi
        projectChildObject1.append("gaRoi", createDivideObject("$gaTransactionRevenue", "$spent", 0L));

        List<DBObject> aggregateObjects = new ArrayList<DBObject>();
        aggregateObjects.add(new BasicDBObject("$match", matchChildObject1));
        aggregateObjects.add(new BasicDBObject("$group", groupChildObject1));
        aggregateObjects.add(new BasicDBObject("$project", projectChildObject1));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateObjects));
        long start = System.currentTimeMillis();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY_90DAYS).aggregate(aggregateObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        JSONArray result = new JSONArray();
        try {
            while (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                result.add(JSON.parse(JSON.toJSONString(dbObject)));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return result;
    }

    /**
     * 创建除法对象
     *
     * @param dividedField 被除数字段
     * @param divideField  除数字段
     * @param defaultValue 除数为0时默认返回的值
     * @return 创建好的对象
     */
    private BasicDBObject createDivideObject(String dividedField, String divideField, Long defaultValue) {
        return new BasicDBObject("$cond",
                new Object[]{
                        new BasicDBObject("$eq", new Object[]{divideField, 0}),
                        defaultValue,
                        new BasicDBObject("$divide", new String[]{dividedField, divideField})
                });
    }

    /**
     * 创建除法对象
     *
     * @param dividedField 被除数字段
     * @param divideField  除数字段
     * @param defaultField 除数为0时默认返回的结果字段
     * @return 创建好的对象
     */
    private BasicDBObject createDivideObject(String dividedField, String divideField, String defaultField) {
        return new BasicDBObject("$cond",
                new BasicDBObject()
                        .append("if", new BasicDBObject("$eq", new Object[]{divideField, 0}))
                        .append("then", defaultField)
                        .append("else", new BasicDBObject("$divide", new String[]{dividedField, divideField})));
    }
}
